package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.LearnVideoPlayBean;
import com.atguigu.func.DimAsyncFunction;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: DwsLearnVideoPlay
 * Package: com.atguigu.app.dws
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/17 11:03
 * @Version 1.0
 */
public class DwsLearnVideoPlayWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.读取kafka dwd_traffic_play_log主题数据
        DataStreamSource<String> kafkaSource = env.fromSource(KafkaUtil.getKafkaSource("dwd_traffic_play_log", "dws_video_play"), WatermarkStrategy.noWatermarks(), "kafka-source");

        //TODO 3.转换为JavaBean对象
        SingleOutputStreamOperator<LearnVideoPlayBean> mapDs = kafkaSource.map(new MapFunction<String, LearnVideoPlayBean>() {
            @Override
            public LearnVideoPlayBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                return LearnVideoPlayBean.builder()
                        .PlayTime(jsonObject.getJSONObject("appVideo").getLong("play_sec"))
                        .videoId(jsonObject.getJSONObject("appVideo").getString("video_id"))
                        .uid(jsonObject.getJSONObject("common").getString("uid"))
                        .ts(jsonObject.getLong("ts"))
                        .PlayCount(1L)
                        .build();
            }
        });

        //TODO 4.提取时间戳生成WaterMark
        SingleOutputStreamOperator<LearnVideoPlayBean> mapDsWithWm = mapDs.assignTimestampsAndWatermarks(WatermarkStrategy.<LearnVideoPlayBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<LearnVideoPlayBean>() {
                    @Override
                    public long extractTimestamp(LearnVideoPlayBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));


        //TODO 5.按uid分组去重
        SingleOutputStreamOperator<LearnVideoPlayBean> keyByUidDs = mapDsWithWm.keyBy(LearnVideoPlayBean::getUid)
                .map(new RichMapFunction<LearnVideoPlayBean, LearnVideoPlayBean>() {
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("last-state", String.class);

                        //设置TTL
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();
                        stateDescriptor.enableTimeToLive(ttlConfig);

                        valueState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public LearnVideoPlayBean map(LearnVideoPlayBean value) throws Exception {
                        //取出状态数据
                        String lastDt = valueState.value();
                        String curDt = DateFormatUtil.toDate(value.getTs());
                        long uct = 0L;
                        if (lastDt == null || !lastDt.equals(curDt)) {
                            uct = 1L;
                            valueState.update(curDt);
                        }
                        value.setUidCount(uct);

                        return value;
                    }
                });

        //TODO 6.关联dim_video_info 拿到chapter_id 再关联 dim_chapter_info 拿到 chapterName
        SingleOutputStreamOperator<LearnVideoPlayBean> WithCIDs = AsyncDataStream.unorderedWait(keyByUidDs, new DimAsyncFunction<LearnVideoPlayBean>("DIM_VIDEO_INFO") {
                    @Override
                    public String getKey(LearnVideoPlayBean input) throws Exception {
                        return input.getVideoId();
                    }

                    @Override
                    public void join(LearnVideoPlayBean input, JSONObject dimInfo) throws Exception {
                        input.setChapterId(dimInfo.getString("CHAPTER_ID"));

                    }
                }
                , 1, TimeUnit.MINUTES);
        SingleOutputStreamOperator<LearnVideoPlayBean> WithCNDs = AsyncDataStream.unorderedWait(WithCIDs, new DimAsyncFunction<LearnVideoPlayBean>("DIM_CHAPTER_INFO") {
                    @Override
                    public String getKey(LearnVideoPlayBean input) throws Exception {
                        return input.getChapterId();
                    }

                    @Override
                    public void join(LearnVideoPlayBean input, JSONObject dimInfo) throws Exception {
                        input.setChapterName(dimInfo.getString("CHAPTER_NAME"));

                    }
                }
                , 1, TimeUnit.MINUTES);

        //TODO 6.分组开窗聚合
        SingleOutputStreamOperator<LearnVideoPlayBean> resultDs = WithCNDs.keyBy(LearnVideoPlayBean::getChapterName)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<LearnVideoPlayBean>() {
                    @Override
                    public LearnVideoPlayBean reduce(LearnVideoPlayBean value1, LearnVideoPlayBean value2) throws Exception {
                        value1.setPlayTime(value1.getPlayTime() + value2.getPlayTime());
                        value1.setPlayCount(value1.getPlayCount() + value2.getPlayCount());
                        value1.setUidCount(value1.getUidCount()+ value2.getUidCount());
                        return value1;
                    }
                }, new WindowFunction<LearnVideoPlayBean, LearnVideoPlayBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<LearnVideoPlayBean> input, Collector<LearnVideoPlayBean> out) throws Exception {
                        LearnVideoPlayBean next = input.iterator().next();
                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        out.collect(next);
                    }
                });
        resultDs.print("resultDs>>>>>");
       resultDs.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_learn_video_play_window values(?,?,?,?,?,?,?)"));
        env.execute();
    }
}
